package com.jscloud.spark.visitcitycount

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object FindIp {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("FindIpWithVisitCitys").setMaster("local[*]")
    val sparkContext = new SparkContext(conf)
    sparkContext.setLogLevel("WARN")

    //读取 http.log 将 IP 地址转换成 long 类型的数据
    val ipLogRDD: RDD[String] = sparkContext.textFile("file:///D:\\JSProjects\\jsCloud-bigdata-app\\datas\\http.log")
      .map(_.split("\\|")(1))

    val ipInfoRDD: RDD[Ip] = sparkContext.textFile("file:///D:\\JSProjects\\jsCloud-bigdata-app\\datas\\ip.dat")
      .map {
        case line: String => {
          val strSplit: Array[String] = line.split("\\|")
          Ip(strSplit(0), strSplit(1), strSplit(7))
        }

      }

    //ip所属地的广播
    val brIpInfo: Broadcast[Array[(Long, Long, String)]] = sparkContext.broadcast(ipInfoRDD.map(x => (ip2Long(x.startIp), ip2Long(x.endIp), x.address)).collect())

    ipLogRDD.map(x => {
      val index: Int = binarySearch(brIpInfo.value, ip2Long(x))
      if (index != -1) {
        brIpInfo.value(index)._3
      } else {
        "NULL"
      }
    }).map(x => (x, 1))
      .reduceByKey(_ + _)
      .map(x => s"城市：${x._1},访问量：${x._2}")
      .saveAsTextFile("file:///D:\\JSProjects\\jsCloud-bigdata-app\\datas\\out_result")


    sparkContext.stop()

  }

  def ip2Long(ip: String): Long = {
    val fragments: Array[String] = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until (fragments.length)) {
      ipNum = fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }

  //二分法匹配ip规则
  def binarySearch(lines: Array[(Long, Long, String)], ip: Long): Int = {
    var low = 0
    var high = lines.length - 1
    while (low <= high) {
      val middle = (low + high) / 2
      if ((ip >= lines(middle)._1) && (ip <= lines(middle)._2))
        return middle
      if (ip < lines(middle)._1)
        high = middle - 1
      else {
        low = middle + 1
      }
    }
    -1
  }

}

//定义一个样例类
case class Ip(startIp: String, endIp: String, address: String)